package com.doit.day02;

import com.alibaba.fastjson.JSON;
import com.doit.day01.MyPartitioner;
import org.apache.commons.lang3.RandomStringUtils;
import org.apache.commons.lang3.RandomUtils;
import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.ProducerConfig;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.apache.kafka.common.serialization.StringSerializer;

import java.util.Properties;
import java.util.Random;

/**
 * 需求：写一个生产者，不断的去生产用户行为数据，写入到kafka的一个topic中  ==>先造一部分数据，然后不断的往kafka中写数据(生产者)
 * 生产的数据格式：  造数据
 * {"guid":1,"eventId":"pageview","timestamp":1637868346789}  isNew = 1  ==》 fastjson  ==》javabean 创建对象  格式化成json串
 * {"guid":1,"eventId":"addcard","timestamp":1637868347625}   isNew = 0
 * {"guid":2,"eventId":"collect","timestamp":16378683463219}
 * {"guid":3,"eventId":"paid","timestamp":16378683467829}
 * ......
 * 再写一个消费者，不断的从kafka中消费上面的用户行为数据，做一个统计  ==》poll for(具体的逻辑)
 * 1.每5s输出一次当前来了多少用户(去重)  uv   每5s输出一次==》 任务调度器 Timer
 * 2.将每条数据添加一个字段来标识，如果这个用户的id是第一次出现，那么就标注1，否则就是0  ==》 判断这个数之前有没有出现过
 */
public class _03_生产者消费者练习题 {
    public static void main(String[] args) throws InterruptedException {
        Properties props = new Properties();
        //这三个参数是必须配置的
        props.setProperty(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG,"linux01:9092,linux02:9092,linux03:9092");
        props.setProperty(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG,StringSerializer.class.getName());
        props.setProperty("value.serializer",StringSerializer.class.getName());
        props.setProperty(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG,StringSerializer.class.getName());
        //选配的
        //kafka生产者应答机制
        props.setProperty("acks","-1");
        //创建生产者对象
        KafkaProducer<String, String> producer = new KafkaProducer<String, String>(props);
        EventLog eventLog = new EventLog();

        while (true){
            eventLog.setEventId(RandomStringUtils.randomAlphabetic(10));
            eventLog.setGuid(RandomUtils.nextInt(10000,100000));
            eventLog.setTimestamp(System.currentTimeMillis());
            //对象有了，值也设置了,将对象变成Json串
            String json = JSON.toJSONString(eventLog);
            ProducerRecord<String, String> record = new ProducerRecord<>("event-log", json);
            producer.send(record);
            producer.flush();
            Thread.sleep(RandomUtils.nextInt(10,200));
        }

    }
}
